package study.rocketmq.origin.orderly;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

/***
 * TODO 顺序消息消费
 *  setPullBatchSize setConsumeMessageBatchMaxSize 仍然可以分批处理。同一队列是同步执行的。
 *  setConsumeThreadMax 和  setConsumeThreadMin可以设置，不过同一队列中，同一时间只能有一个线程进行。不同队列之间可以并发。
 *
 */
public class OrderlyConsumer {
    static Logger logger = LoggerFactory.getLogger(OrderlyConsumer.class);

    public static final String NAMESRC_ADDR = "120.25.73.46:9876";
    public static void main(String[] args) throws Exception {
        /**
         * TODO
         */
        //参数为 消费者组。
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer("Consumer-Group-01");

        defaultMQPushConsumer.setNamesrvAddr(NAMESRC_ADDR);
        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        // 支持多个使用||分割。如 TagA||TagB
        // 如果send中有多个Tag，比如TagA，TagB，消费者只监听了一个TagA，那么TagB也可能被拉取到消费者中，但是会忽略TagB的消息（等于消息丢失了）。
        defaultMQPushConsumer.subscribe("topic-orderly-01","*");

        /**TODO 一次性拉取多少消息, 默认32  */
        defaultMQPushConsumer.setPullBatchSize(6);
        /**TODO 一次性拉取PullBatchSize后，分批处理每批的个数 - consumeMessage中List<MessageExt>的个数, 默认1
         *  注意：批次可以并发处理，并发线程见ConsumeThreadMax和setConsumeThreadMin设置
         **/
        defaultMQPushConsumer.setConsumeMessageBatchMaxSize(4);
        /**TODO 最大线程数和最小线程数  默认20  */
        defaultMQPushConsumer.setConsumeThreadMax(20);
        defaultMQPushConsumer.setConsumeThreadMin(20);

        defaultMQPushConsumer.setMaxReconsumeTimes(1);
        // 集群模式还是广播模式
        defaultMQPushConsumer.setMessageModel(MessageModel.CLUSTERING);

        defaultMQPushConsumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {

                for(MessageExt me : msgs) {
                    String keys = me.getKeys();
                    String topic = me.getTopic();
                    String tags = me.getTags();
                    int QueueId = context.getMessageQueue().getQueueId();

                    try {
                        String msgBody = new String(me.getBody(), RemotingHelper.DEFAULT_CHARSET);
                        // 消费逻辑 ---->
                        if (keys.equals("key-2")) {
                            logger.info(keys + "-开始消费 -- > msgs.size =" + msgs.size());
                            int a = 1 / 0;
                        }
                        logger.info(keys + "-消费成功- topic: " + topic + ",tags: " + tags );
                    } catch (Exception e) {
                        e.printStackTrace();
                        /** TODO 重试的次数（不包括第一次失败） */
                        int recousumeTimes = me.getReconsumeTimes();
                        /** TODO 重发3次就不重试了 */
                        if (recousumeTimes == 3) {
                            //		记录日志....
                            //  	做补偿处理
                            /** TODO 设置返回成功，就不重试了。 */
                            logger.error(keys + "消费失败-已经重试了【" + recousumeTimes + "】次，不再重试了 -- >");
                            return ConsumeOrderlyStatus.SUCCESS;
                        } else {
                            logger.error(keys + "消费失败-准备重试- 已重试次数: " + recousumeTimes);
                        }
                        /** TODO RabbitMQ会自动重试，共1次，重试时间间隔如下：
                         *     10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
                         *     ps：延迟消息多了两档：1s 5s，其余一样。
                         *         messageDelayLevel = 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
                         * */
                        return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                    }
                }
                return ConsumeOrderlyStatus.SUCCESS;


            }
        });
        defaultMQPushConsumer.start();
        System.err.println("consumer start...");
    }
}
